package com.gengzp.mq.server.manager.impl;

import com.gengzp.mq.server.facade.MqMessageFacadeService;
import com.gengzp.mq.server.manager.MqObsoleteMessageManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

import static com.gengzp.mq.server.constants.MqThreadNameConstants.MQ_MESSAGE_OBSOLETE_CLEAR_THREAD_NAME;
import static com.gengzp.mq.server.constants.MqWaitTimeConstants.CLEAR_OBSOLETE_MESSAGE_INTERVAL;

/**
 * @ClassName MqObsoleteMessageManagerImpl
 * @Description
 * @Author gengzp
 * @Date 2025/9/4 14:55
 */
@Component
public class MqObsoleteMessageManagerImpl implements MqObsoleteMessageManager {

    private static final Logger logger = LoggerFactory.getLogger(MqObsoleteMessageManagerImpl.class);

    @Autowired
    private MqMessageFacadeService mqMessageFacadeService;

    /**
     * 废弃消息清理线程
     */
    private static volatile Thread OBSOLETE_MESSAGE_CLEAR_THREAD = null;

    @Override
    public void createObsoleteMessageJob() {
        if (OBSOLETE_MESSAGE_CLEAR_THREAD != null && OBSOLETE_MESSAGE_CLEAR_THREAD.isAlive()) {
            return;
        }

        OBSOLETE_MESSAGE_CLEAR_THREAD = new Thread(() -> {
            while (true) {
                // 查询所有废弃消息
                logger.info("============== 开始执行历史消息清理任务 ==============");
                try {
                    List<String> obsoleteMessageIds = mqMessageFacadeService.searchAllObsoleteMessageIds();
                    if (obsoleteMessageIds != null && !obsoleteMessageIds.isEmpty()) {
                        for (int i = 0; i < obsoleteMessageIds.size(); i = i + 200) {
                            mqMessageFacadeService.removeMessagesByIds(
                                    obsoleteMessageIds.subList(i, Math.min(i + 200, obsoleteMessageIds.size())));
                        }
                        logger.info("清理过时消息 {} 条", obsoleteMessageIds.size());
                    } else {
                        logger.info("无过时消息, 无需清理");
                    }
                    logger.info("============== 历史消息清理任务执行完成 ==============");
                } catch (Exception e) {
                    logger.error("============== 历史消息清理任务执行失败: {} ==============", e.getMessage(), e);
                }

                try {
                    Thread.sleep(CLEAR_OBSOLETE_MESSAGE_INTERVAL);
                } catch (Exception e) {
                    logger.error("废弃消息清理线程休眠异常: {}", e.getMessage());
                }
            }
        }, MQ_MESSAGE_OBSOLETE_CLEAR_THREAD_NAME);

        // 启动线程
        OBSOLETE_MESSAGE_CLEAR_THREAD.start();
        logger.info("废弃消息清理线程 {} 启动成功", MQ_MESSAGE_OBSOLETE_CLEAR_THREAD_NAME);
    }

}
